package com.ww.exercise.coroutine.mid.q12

import com.ww.coroutines.launch
import com.ww.coroutines.scope.coroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.channels.trySendBlocking
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit
import java.util.concurrent.CopyOnWriteArrayList
import kotlin.random.Random

// 1.定义任务优先级枚举
enum class TaskPriority { HIGH, LOW }

// 2.任务数据类：包含优先级、唯一ID、创建时间（用于计算耗时和超时）
data class Task(
    val priority: TaskPriority,
    val id: String,
    val createTime: Long = System.currentTimeMillis()
)

// 3.任务处理结果：记录任务、耗时、是否超时、处理状态
data class TaskResult(
    val task: Task,
    val processTimeMs: Long,
    val isTimeout: Boolean,
    val status: String
)

// 4.优先级任务调度器：核心逻辑
class PriorityTaskScheduler(
    private val maxConCurrency: Int = 20, // 总最大并发数
    private val highPriorityTimeoutMs: Long = 10000 // 高优先级任务超时时间（10s）
) {
    // 高优先级任务通道（无界缓冲，避免生产者阻塞）
    private val highPriorityChannel: Channel<Task> = Channel(Channel.UNLIMITED)

    // 低优先级任务通道（无界缓冲）
    private val lowPriorityChannel: Channel<Task> = Channel(Channel.UNLIMITED)

    // 控制总并发的信号量
    private val concurrencySemaphore = Semaphore(maxConCurrency)

    // 线程安全的结果列表（存储所有任务处理结果）
    private val taskResults = CopyOnWriteArrayList<TaskResult>()

    // 提交任务到对应通道
    fun submitTask(task: Task) {
        when (task.priority) {
            TaskPriority.HIGH -> highPriorityChannel.trySendBlocking(task)
            TaskPriority.LOW -> lowPriorityChannel.trySendBlocking(task)
        }
    }

    // 启动调度器：优先消费高优先级任务
    suspend fun startScheduler() = coroutineScope {
        // 调度协程：循环接收并处理任务
        launch(Dispatchers.Default) {
            while (true) {
                // 核心逻辑：先消费高优先级通道，无数据再消费低优先级通道
                val task = receiveHighPriorityTask() ?: receiveLowPriorityTask() ?: break
                // 处理任务
                launch { processTask(task) }
            }
        }

        // 等待所有通道关闭且任务处理完成
        launch {
            highPriorityChannel.consumeEach {  }
            lowPriorityChannel.consumeEach {  }

            while (concurrencySemaphore.availablePermits < maxConCurrency) {
                delay(100)
            }
            // 关闭调度器时打印统计报告
            printStatisticsReport()
        }
    }

    // 关闭所有通道（停止接收新任务）
    fun stopScheduler() {
        highPriorityChannel.close()
        lowPriorityChannel.close()
    }

    // 优先接收高优先级任务，非阻塞
    private fun receiveHighPriorityTask(): Task? {
        return highPriorityChannel.tryReceive().getOrNull()
    }

    // 接收低优先级任务（阻塞，通道关闭返回null）
    private suspend fun receiveLowPriorityTask(): Task? {
        return lowPriorityChannel.receiveCatching().getOrNull()
    }

    // 处理单个任务：控制并发、记录耗时、判断超时
    private suspend fun processTask(task: Task) {
        concurrencySemaphore.withPermit {
            val processStartTime = System.currentTimeMillis()
            var isTimeout = false
            var status = "处理完成"

            try {
                // 模拟任务处理耗时
                // 高优先级任务：短耗时（100-1000ms），低优先级任务：长耗时（1000-5000ms）
                val processDelayMs = when (task.priority) {
                    TaskPriority.HIGH -> Random.nextLong(100, 1000)
                    TaskPriority.LOW -> Random.nextLong(1000, 5000)
                }

                // 高优先级任务检查超时，处理前先判断是否已超时（从创建到开始处理超过10s）
                if (task.priority == TaskPriority.HIGH) {
                    val waitTimeMs = processStartTime - task.createTime
                    if (waitTimeMs > highPriorityTimeoutMs) {
                        isTimeout = true
                        status = "已超时（等待时间：${waitTimeMs}ms）"
                        println("⚠高优先级任务${task.id}超时，未执行")
                        return
                    }
                }

                // 执行任务（模拟耗时）
                delay(processDelayMs)
                val processTimeMs = System.currentTimeMillis() - processStartTime
                println("✅${task.priority.name}任务${task.id}处理完成，耗时${processTimeMs}ms")

                // 记录结果
                taskResults.add(
                    TaskResult(
                        task = task,
                        processTimeMs = processTimeMs,
                        isTimeout = isTimeout,
                        status = status
                    )
                )
            } catch (e: Exception) {
                val processTimeMs = System.currentTimeMillis() - processStartTime
                status = "处理失败：${e.message}"
                taskResults.add(
                    TaskResult(
                        task = task,
                        processTimeMs = processTimeMs,
                        isTimeout = isTimeout,
                        status = status
                    )
                )
                println("❌${task.priority.name}任务${task.id}处理失败${e.message}")
            }
        }
    }

    // 生成统计报告：验证高优先级任务耗时优势
    private fun printStatisticsReport() {
        val highResults = taskResults.filter { it.task.priority == TaskPriority.HIGH }
        val lowResults = taskResults.filter { it.task.priority == TaskPriority.LOW }
        val highTimeoutCount = highResults.count { it.isTimeout }

        println("\n" + "=".repeat(80))
        println("📊 优先级任务调度统计报告")
        println("=".repeat(80))
        println("1. 任务总量：${taskResults.size}（高优先级：${highResults.size}，低优先级：${lowResults.size}）")
        println("2. 高优先级任务：")
        println("   - 平均处理耗时：${if (highResults.isNotEmpty()) highResults.sumOf { it.processTimeMs } / highResults.size else 0}ms")
        println(
            "   - 超时数量：${highTimeoutCount}（超时率：${
                if (highResults.isNotEmpty()) "%.2f%%".format(
                    highTimeoutCount * 100.0 / highResults.size
                ) else "0.00%"
            }）"
        )
        println("3. 低优先级任务：")
        println("   - 平均处理耗时：${if (lowResults.isNotEmpty()) lowResults.sumOf { it.processTimeMs } / lowResults.size else 0}ms")
        println("4. 总并发控制：最大${maxConCurrency}个协程，无超并发情况")
        println("=".repeat(80))
    }
}

fun main() = runBlocking {
    // 初始化调度器（最大并发20，高优先级超时10s）
    val scheduler = PriorityTaskScheduler(maxConCurrency = 20)

    // 模拟任务生产者：同时产生高、低优先级任务
    launch(Dispatchers.IO) {
        // 高优先级任务生产者：每100ms产生1个，持续10s
        repeat(100) {
            val task = Task(TaskPriority.HIGH, "H-${it + 1}")
            scheduler.submitTask(task)
            delay(100)
        }
        println("\n\uD83D\uDCE4高优先级任务生产完成（共100个）")
    }

    launch(Dispatchers.IO) {
        // 低优先级任务生产者：每500ms产生1个，持续15s
        repeat(30) {
            val task = Task(TaskPriority.LOW, "L-${it + 1}")
            scheduler.submitTask(task)
            delay(500)
        }
        println("\n\uD83D\uDCE4低优先级任务生产完成（共30个）")
    }

    // 启动调度器，等待所有任务处理完成
    scheduler.startScheduler()

    // 所有任务处理完成后，停止调度器
    scheduler.stopScheduler()
    println("\n调度系统已停止")
}